package com.jacquesh;

import com.jacquesh.model.CoffeeShop;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.text.SimpleDateFormat;

public class FlinkDemo0012 {

    public static void main(String[] args) throws Exception {
        SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd");

        StreamExecutionEnvironment environment =
                StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> stream = environment
                .socketTextStream("192.168.88.180", 8888);
        final OutputTag<CoffeeShop> outputTag = new OutputTag<CoffeeShop>("output"){};
        SingleOutputStreamOperator dataStream = stream.
                flatMap(new RichFlatMapFunction<String, CoffeeShop>() {
                    @Override
                    public void flatMap(String s, Collector<CoffeeShop> collector) {
                        try {
                            String[] sp = s.split(",");
                            collector.collect(CoffeeShop.builder()
                                    .id(Long.parseLong(sp[0]))
                                    .date(format.parse(sp[1]))
                                    .sales(Integer.parseInt(sp[2]))
                                    .build());
                        } catch (Exception ex) {
                            System.out.println(ex);
                        }

                    }
                })        /** 依据id分组*/
                .keyBy(CoffeeShop::getId)
                .process(new KeyedProcessFunction<Long, CoffeeShop, String>() {

                    @Override
                    public void processElement(
                            CoffeeShop coffeeShop,
                            Context context,
                            Collector<String> collector)  {
                        try{
                            /**当前处理时间*/
                            long currentTime=  context.timerService()
                                    .currentProcessingTime();
                            if(coffeeShop.getSales()>100){
                                 long timerTime = currentTime + 3 * 1000;
                                 context.timerService()
                                         .registerProcessingTimeTimer(timerTime);
                            }else{
                                collector.collect(coffeeShop
                                        .toString()+"未到达100");
                            }

                        }
                        catch (Exception ex){
                            System.out.println(ex);
                        }
                    }
                    /**设置时钟时间*/
                    @Override
                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out)  {
                        out.collect( ctx.getCurrentKey()+"销量到达：100");
                    }
                });
          dataStream.print("dataStream").setParallelism(1);

        environment.execute("process demo");
    }

}



